package com.lrj.datastream.transform.udf;

import com.lrj.datastream.domain.ClickEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * UDF的实现方式
 * - 类继承相关算子Function或者实现Function或者RichFunction
 * - 匿名内类
 * - Lamda表达式
 * 需要注意的是使用lamda表达式,如果是带泛型的,需要显式的returns指明类型
 * 否则无法推导出类型
 *
 * @author lrj
 * @date 2022/3/28 17:26
 */
public class TestRichFunc {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /* 一个任务一个open close
=============open: 4 =============
=============open: 2 =============
=============open: 3 =============
=============open: 6 =============
=============open: 5 =============
=============open: 1 =============
=============open: 7 =============
=============open: 0 =============
2> 6
1> 16
8> 6
------------------close------------------
------------------close------------------
------------------close------------------
------------------close------------------
------------------close------------------
------------------close------------------
------------------close------------------
------------------close------------------
        */
        DataStreamSource<ClickEvent> stream = env.fromElements(
                new ClickEvent("Jack", "./home", 1000L),
                new ClickEvent("Mary", "./product?id=100", 2000L),
                new ClickEvent("Alice", "./cart", 3000L)
        );
        // 1.基于类,类的方式可以传递相关参数给构造函数
        stream.map(new MyRichMapper()).print();
        // 2.基于匿名内部类
        stream.map(new MapFunction<ClickEvent, Object>() {
            @Override
            public Object map(ClickEvent value) throws Exception {
                return value.getUrl();
            }
        }).print("匿名内部类");
        // 3.基于lamda
        stream.map(ClickEvent::getUrl).print("lamda表达式");
        env.execute(TestRichFunc.class.getName());
    }

    private static class MyRichMapper extends RichMapFunction<ClickEvent, Integer> {
        @Override
        public void open(Configuration parameters) throws Exception {
            RuntimeContext runtimeContext = getRuntimeContext();
            int indexOfThisSubtask = runtimeContext.getIndexOfThisSubtask();
            System.out.println("=============open: " + indexOfThisSubtask + " =============");
        }

        @Override
        public Integer map(ClickEvent value) throws Exception {
            return value.getUrl().length();
        }

        @Override
        public void close() throws Exception {
            System.out.println("------------------close------------------");
        }
    }
}
